package com.huawei.bigdata.kafka.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Component
public class ProducerConfig {
//    @Value("${conf.kafka.test.topic}")
//    private String topic;
    @Value("${conf.kafka.test.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${conf.kafka.test.security-protocol}")
    private String securityProtocol;
    @Value("${conf.kafka.test.kerberos-domain-name}")
    private String kerberosDomainName;
    @Value("${conf.kafka.test.security}")
    private Boolean security;

    // Broker地址列表
    private final static String BOOTSTRAP_SERVER = "bootstrap.servers";

    // 客户端ID
    private final static String CLIENT_ID = "client.id";

    // Key序列化类
    private final static String KEY_SERIALIZER = "key.serializer";

    // Value序列化类
    private final static String VALUE_SERIALIZER = "value.serializer";

    // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
    private final static String SECURITY_PROTOCOL = "security.protocol";

    // 服务名
    private final static String SASL_KERBEROS_SERVICE_NAME = "sasl.kerberos.service.name";

    // 域名
    private final static String KERBEROS_DOMAIN_NAME = "kerberos.domain.name";

    // 分区类名
    private final static String PARTITIONER_NAME = "partitioner.class";


    @Bean
    public Properties initProperties() {

        Properties props = new Properties();
        ////////
        props.put("batch.size", 163840);
        props.put("linger.ms", 1000);
        props.put("buffer.memory", 67108864);
        props.put("compression.type", "lz4");
        props.put("max.request.size", "524288000");   // 500M
        /////////
        // Broker地址列表
        props.put(BOOTSTRAP_SERVER, bootstrapServers);
        // 客户端ID
        props.put(CLIENT_ID, "DemoProducer");
        // Key序列化类
        props.put(KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer");
        // Value序列化类
        props.put(VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer");
        if (security) {
            // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
            props.put(SECURITY_PROTOCOL, securityProtocol);
        }

        // 服务名
        props.put(SASL_KERBEROS_SERVICE_NAME, "kafka");
        // 域名
        props.put(KERBEROS_DOMAIN_NAME, kerberosDomainName);
        // 分区类名
        props.put(PARTITIONER_NAME, "com.huawei.bigdata.kafka.example.SimplePartitioner");
        return props;
    }


}
